package org.tao.lightningmq.broker.event.spi.listenr;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.core.CommitLogMMapFileModel;
import org.tao.lightningmq.broker.event.model.TimeWheelEvent;
import org.tao.lightningmq.broker.model.TxMessageAckModel;
import org.tao.lightningmq.broker.timewheel.DelayMessageDTO;
import org.tao.lightningmq.broker.timewheel.SlotStoreTypeEnum;
import org.tao.lightningmq.broker.timewheel.TimeWheelSlotModel;
import org.tao.lightningmq.common.dto.*;
import org.tao.lightningmq.common.enums.BrokerResponseCode;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

/**
 * @Author lht
 * @date 2025/1/22 - 15:34
 * @description:
 */
public class TimeWheelListener implements Listener<TimeWheelEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(TimeWheelListener.class);

    @Override
    public void onReceive(TimeWheelEvent event) throws Exception {
        List<TimeWheelSlotModel> timeWheelSlotModelList = event.getTimeWheelSlotModelList();
        for (TimeWheelSlotModel timeWheelSlotModel : timeWheelSlotModelList) {
            if (SlotStoreTypeEnum.MESSAGE_RETRY_DTO.getClazz().equals(timeWheelSlotModel.getStoreType())) {
                MessageRetryDTO messageRetryDTO = (MessageRetryDTO) timeWheelSlotModel.getData();
                this.messageRetryHandler(messageRetryDTO);
            } else if (SlotStoreTypeEnum.DELAY_MESSAGE_DTO.getClazz().equals(timeWheelSlotModel.getStoreType())) {
                MessageDTO messageDTO = (MessageDTO) timeWheelSlotModel.getData();
                LOGGER.info("延迟消息重新入commitLog:{}", JSON.toJSONString(messageDTO));
                CommonCache.getCommitLogAppendHandler().appendMsg(messageDTO, event);
            } else if (SlotStoreTypeEnum.TX_MESSAGE_DTO.getClazz().equals(timeWheelSlotModel.getStoreType())) {
                TxMessageDTO txMessageDTO = (TxMessageDTO) timeWheelSlotModel.getData();
                TxMessageAckModel txMessageAckModel = CommonCache.getTxMessageAckModelMap().get(txMessageDTO.getMsgId());
                if (txMessageAckModel == null) {
                    LOGGER.warn("txMessageAckModel is already been ack");
                    continue;
                }
                TxMessageCallbackReqDTO txMessageCallbackReqDTO = new TxMessageCallbackReqDTO();
                txMessageCallbackReqDTO.setMessageDTO(txMessageAckModel.getMessageDTO());
                TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.TX_CALLBACK_MSG.getCode(), JSON.toJSONBytes(txMessageCallbackReqDTO));
                txMessageAckModel.getChannelHandlerContext()
                        .writeAndFlush(tcpMsg)
                        .addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (future.isSuccess()) {
                            LOGGER.info("txMessageCallbackReqDTO send success");
                            DelayMessageDTO delayMessageDTO = new DelayMessageDTO();
                            delayMessageDTO.setData(txMessageDTO);
                            delayMessageDTO.setSlotStoreType(SlotStoreTypeEnum.TX_MESSAGE_DTO);
                            delayMessageDTO.setNextExecuteTime(System.currentTimeMillis() + 3 *1000);
                            delayMessageDTO.setDelay(3);
                            CommonCache.getTimeWheelModelManager().addDelayMessage(delayMessageDTO);
                        } else {
                            LOGGER.warn("txMessageCallbackReqDTO send fail");
                        }
                    }
                });
            }
        }
    }

    private void messageRetryHandler(MessageRetryDTO messageRetryDTO) {
        CommitLogMMapFileModel commitLogMMapFileModel = CommonCache.getCommitLogMMapFileModelManager().get(messageRetryDTO.getTopic());
        ConsumeMsgCommitLogDTO consumeMsgCommitLogDTO = commitLogMMapFileModel.readContent(messageRetryDTO.getSourceCommitLogOffset(), messageRetryDTO.getSourceCommitLogSize());
        byte[] commitLogBody = consumeMsgCommitLogDTO.getBody();
        LOGGER.info("扔到重试topic的数据：{}", new String(commitLogBody));
        MessageDTO messageDTO = new MessageDTO();
        messageDTO.setBody(commitLogBody);
        messageDTO.setTopic("retry%" + messageRetryDTO.getConsumeGroup());
        messageDTO.setQueueId(ThreadLocalRandom.current().nextInt(3));
        messageDTO.setCurrentRetryTimes(messageRetryDTO.getCurrentRetryTimes());
        LOGGER.info("重试消息投递：{}", messageDTO);
        try {
            CommonCache.getCommitLogAppendHandler().appendMsg(messageDTO);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
